package com.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object CreateRDD {



  def main(args: Array[String]): Unit = {
    //1.准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    //设置分区数
    sparkConf.set("spark.default.parallelism","2")
    //由于我这台内存不足。所以我指定Spark内存
    //sparkConf.set("spark.testing.memory","471859200")
    //2.创建上下文 加载配置信息
    val sparkContext = new SparkContext(sparkConf)

    //3.创建一个方法 这个方法用于在内存中创建RDD
    //memoryCreatRDD(sparkContext)
    //4.从外部存储（文件）创建RDD
    fileCreateRDD(sparkContext)
  }

  def memoryCreatRDD(sparkContext: SparkContext): Unit ={
    //创建RDD 在内存中
    val rdd1:RDD[Int] = sparkContext.parallelize( 1 to 10)
    //第二个参数 是分区数量
    val rdd2:RDD[Int] = sparkContext.parallelize( 1 to 10,3)
    //获取分区数量
    println(rdd1.partitions.length)//2
    println(rdd2.getNumPartitions)//3
    //将结果保存的磁盘当中
    //rdd1.saveAsTextFile("output")

    //内存中创建RDD第二种方式 也是后面最常用
    val rdd3:RDD[Int] = sparkContext.makeRDD(1 to 10,4)
    println(rdd3.getNumPartitions) // 4
    //将rdd3进行迭代
    rdd3.collect().foreach(println)
    rdd3.saveAsTextFile("output")
    //关闭资源
    sparkContext.stop()

  }

  def fileCreateRDD(sparkContext: SparkContext): Unit ={
    //从文件中创建RDD
    val rdd4:RDD[String] = sparkContext.textFile("input/1.txt")
    rdd4.saveAsTextFile("output")
    rdd4.collect().foreach( println )
    //关闭资源
    sparkContext.stop()
  }

}
